
mod simple;
mod general;
mod filter;

/*
PhysicalExpr就是对单个事件数据或者一组事件数据的 判断或处理
*/
use crate::data_frame::{ConfValue, RealValue, ElementValue, DataInstance};
use std::collections::VecDeque;
use crate::logical_plan::expr::{LogicalPlanExpr, Operator};
use async_channel::{Receiver, Sender};
use crate::run_plan::{HandResult, HandResultFail};
use std::ops::Deref;
use std::fmt::Debug;
use async_trait::async_trait;
use crate::physical_plan::general::uuid;
use crate::physical_plan::filter::{gt, ge, and, or};
use crate::physical_plan::simple::set;
use crate::conf_init::{MapResult, ResultErr};

#[derive(Clone,Debug)]
pub enum PhysicalPlanExpr {
    ConfValue(ConfValue),
    RealValue(RealValue),
    ElementValue(ElementValue),
    ElementListValue(Vec<ElementValue>),
}

#[async_trait]
pub trait PhysicalPlanTrait: Debug+Send+Sync{
    fn name(&self)->&str;
    async fn evaluate(&self, args: &[Box<PhysicalPlanExpr>], data: &mut DataInstance) -> HandResult;
}

pub struct PhysicalPlan{
    name: String,
    fn_ptr: fn(&[Box<PhysicalPlanExpr>], data: &mut DataInstance) ->HandResult,
}

impl std::fmt::Debug for PhysicalPlan{
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_tuple("PhysicalPlan")
            .field(&self.name)
            .finish()
    }
}

#[async_trait]
impl PhysicalPlanTrait for PhysicalPlan{
    fn name(&self)->&str{
        return self.name.as_str();
    }
    async fn evaluate(&self, args: &[Box<PhysicalPlanExpr>], data: &mut DataInstance)->HandResult{
        return (self.fn_ptr)(args, data);
    }
}

impl PhysicalPlan{
    fn new(name: &str, fn_ptr: fn(&[Box<PhysicalPlanExpr>],&mut DataInstance)->HandResult)->PhysicalPlan{
        PhysicalPlan{
            name: name.to_string(),
            fn_ptr: fn_ptr,
        }
    }
}

pub fn logicalplan_expr_map_physicalplan_expr(expr: &LogicalPlanExpr)-> MapResult<Box<PhysicalPlanExpr>>{
    match expr{
        LogicalPlanExpr::Literal(value) => {
            return MapResult::Ok(Box::new(PhysicalPlanExpr::RealValue(value.clone())));
        },
        LogicalPlanExpr::LiteralConf(value) => {
            return MapResult::Ok(Box::new(PhysicalPlanExpr::ConfValue(value.clone())));
        },
        LogicalPlanExpr::Column(value) => {
            return MapResult::Ok(Box::new(PhysicalPlanExpr::ElementValue(value.clone())));
        },
        LogicalPlanExpr::Columns(values) => {
            return MapResult::Ok(Box::new(PhysicalPlanExpr::ElementListValue(values.clone())));
        },
        other => {
            return MapResult::Err(ResultErr::new(format!("logicalplan map to physicalplan, not support {:?}",
                                                            other)));
        },
    }
}

pub fn logicalplan_func_map_physicalplan(name: &str)-> MapResult<Box<dyn PhysicalPlanTrait>>{
    match name{
        "gt" => {return MapResult::Ok(Box::new(PhysicalPlan::new(name, gt)));}
        "ge" => {return MapResult::Ok(Box::new(PhysicalPlan::new(name, ge)));}
        "and" => {return MapResult::Ok(Box::new(PhysicalPlan::new(name, and)));}
        "or" => {return MapResult::Ok(Box::new(PhysicalPlan::new(name, or)));}
        "set" => {return MapResult::Ok(Box::new(PhysicalPlan::new(name, set)));}
        "uuid" => {return MapResult::Ok(Box::new(PhysicalPlan::new(name, uuid)));}

        other => {
            return MapResult::Err(ResultErr::new(format!("func {} not support", other)));
        },
    }
}

